package com.hna.face.mqtt;

import java.io.File;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.net.ssl.SSLSocketFactory;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.toolkit.StringUtils;
import com.hna.face.config.SysConfig;
import com.hna.face.security.ssl.SSLUtils;

@Service
public class MqttService {

	private final static Logger log = LoggerFactory.getLogger(MqttService.class);

	private ScheduledExecutorService scheduler;

	private MqttClient client;

	private MqttTopic mqttTopic;

	private MqttConnectOptions options;
	
	@Autowired
	private PushCallback callback;

	@PostConstruct
	public void init() {
		try {
			client = new MqttClient(SysConfig.mqttHost, SysConfig.mqttClientid, new MemoryPersistence());
			connect();
			startReconnect();
		} catch (MqttException e1) {
			log.error("mqtt init", e1);
		}
	}

	private void connect() {
		options = new MqttConnectOptions();
		options.setCleanSession(false);
		log.info(SysConfig.mqttUserName);
		log.info(SysConfig.mqttPassword);

		options.setUserName(SysConfig.mqttUserName);
		options.setPassword(SysConfig.mqttPassword.toCharArray());
		// 设置超时时间
		options.setConnectionTimeout(SysConfig.mqttConnectionTimeout);
		// 设置会话心跳时间
		options.setKeepAliveInterval(SysConfig.mqttKeepAliveInterval);
		try {
			SSLUtils sslUtils = new SSLUtils();
			log.info("地址:"+SysConfig.sslPath);
			String sslPath;
			if(StringUtils.isEmpty(SysConfig.sslPath))
			{
				sslPath = this.getClass().getClassLoader().getResource("").getFile()+File.separator+"config"+File.separator;
			}
			else {
				sslPath=SysConfig.sslPath;
			}
			log.info("地址2:"+sslPath);
			/*String sslPath = this.getClass().getClassLoader().getResource("").getFile();
			SSLSocketFactory factory = sslUtils.getSSLSocktet(sslPath+"config/ca.crt",sslPath+"config/client1.crt",
					sslPath+"config/client1.pem", "");*/
			SSLSocketFactory factory = sslUtils.getSSLSocktet(sslPath+"ca.crt",sslPath+"client1.crt",
					sslPath+"client1.pem", "");
			options.setSocketFactory(factory);

			if (callback == null) {
				log.error("callback is null");
			} else {
				log.info("callback is ok");
			}

			client.setCallback(callback);
			client.connect(options);
			
			//订阅消息
			String[] topics = { SysConfig.mqttWill, SysConfig.mqttOnline, SysConfig.mqttData };
			for (String topic : topics) {
				log.info("subscribe " + topic);
				if (client == null) {
					log.error("client is null");
				}
				this.subscribe(2,topic);
				log.info("subscribe is ok");
			}
		} catch (Exception e) {
			log.error("connect:", e);
		}
	}
	// 重新链接
		public void startReconnect() {
			scheduler = Executors.newSingleThreadScheduledExecutor();
			scheduler.scheduleAtFixedRate(new Runnable() {
				public void run() {
					if (!client.isConnected()) {
						try {
							log.info("重新链接");
							client.connect(options);
						} catch (MqttSecurityException e) {
							log.error("MqttSecurityException:", e);
						} catch (MqttException e) {
							log.error("MqttException:", e);
						}
					}
				}
			}, SysConfig.mqttInitialDelay, SysConfig.mqttPeriod, TimeUnit.MILLISECONDS);
		}
	
	public boolean publish(MqttMessage message, String topic) throws MqttPersistenceException, MqttException {
		mqttTopic = client.getTopic(topic);
		message.setRetained(false);
		MqttDeliveryToken token = mqttTopic.publish(message);
		token.waitForCompletion();
		log.info("publishStatus "+token.isComplete());
		return token.isComplete();
	}
	public void subscribe(int qos,String topic) throws MqttException {
		log.info("subscribeTopics " + topic);
		int[] Qos = { qos };
		String[] topic1 = { topic };
		if (client == null) {
			log.error("subscribeClient is null");
		}
		client.subscribe(topic1, Qos);
		log.info("subscribeTopics " + topic+" is ok");
	}
}
